package net.bw.realtime.jtp.dwd.log.function;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import net.bw.realtime.jtp.common.utils.JdbcUtil;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;

/**
 * @author liuyawei
 * @date 2025/5/22
 */
public class JtpLogClickHouseSinkFunction {

    // 页面日志存储到clickhouse
    public static void savePageLog(DataStream<String> stream){

        // 1.转换格式
        SingleOutputStreamOperator<String> jsonStream = stream.map(
                new RichMapFunction<String, String>() {
                    @Override
                    public String map(String value) throws Exception {
                        // 1解析JSON
                        JSONObject object = JSON.parseObject(value);
                        JSONObject commonObject = object.getJSONObject("common");
                        JSONObject pageObject = object.getJSONObject("page");

                        // 2提前字段值
                        JSONObject jsonObject = new JSONObject() ;
                        jsonObject.put("mid", commonObject.getString("mid")) ;
                        jsonObject.put("area_code", commonObject.getString("ar")) ;
                        jsonObject.put("brand", commonObject.getString("ba")) ;
                        jsonObject.put("channel", commonObject.getString("ch")) ;
                        jsonObject.put("is_new", commonObject.getString("is_new")) ;
                        jsonObject.put("model", commonObject.getString("md")) ;
                        jsonObject.put("os", commonObject.getString("os")) ;
                        jsonObject.put("uid", commonObject.getOrDefault("uid", "").toString()) ;
                        jsonObject.put("version_code", commonObject.getString("vc")) ;

                        jsonObject.put("page_id", pageObject.getOrDefault("page_id", "")) ;
                        jsonObject.put("last_page_id", pageObject.getOrDefault("last_page_id", "")) ;
                        jsonObject.put("during_time", pageObject.getLong("during_time")) ;
                        jsonObject.put("sourceType", pageObject.getOrDefault("sourceType", "")) ;
                        jsonObject.put("item_type", pageObject.getOrDefault("item_type", "")) ;
                        jsonObject.put("item", pageObject.getOrDefault("item", "")) ;

                        jsonObject.put("ts", object.getLong("ts")) ;
                        // 3返回JSON字符串
                        return jsonObject.toJSONString();
                    }
                }
        );

        // 2.插入数据
        String[] columns = new String[]{
                "mid", "area_code","brand", "channel", "is_new", "model", "os", "uid", "version_code",
                "page_id", "last_page_id", "during_time", "sourceType", "item_type", "item", "ts"
        };

        // 3.构造SQL
        String upsertSql = "INSERT INTO jtp_app_log.dwd_traffic_page_log(\n" +
                "    mid, area_code, brand, channel, is_new, model, os, uid, version_code,\n" +
                "    page_id, last_page_id, during_time, sourceType, item_type, item,\n" +
                "    ts\n" +
                ") VALUES (\n" +
                "   ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?\n" +
                ")" ;

        // 4.执行
        JdbcUtil.sinkToClickhouseUpsertJson(jsonStream,columns, upsertSql);

    }

    // 启动日志存储clickhouse
    public static void saveStartLog(DataStream<String> stream){

        // 1.转换数据格式
        SingleOutputStreamOperator<String> jsonStream = stream.map(
                new RichMapFunction<String, String>() {
                    @Override
                    public String map(String value) throws Exception {
                        // 1解析JSON
                        JSONObject object = JSON.parseObject(value);
                        JSONObject commonObject = object.getJSONObject("common");
                        JSONObject startObject = object.getJSONObject("start");

                        // 2提前字段值
                        JSONObject jsonObject = new JSONObject() ;
                        jsonObject.put("mid", commonObject.getString("mid")) ;
                        jsonObject.put("area_code", commonObject.getString("ar")) ;
                        jsonObject.put("brand", commonObject.getString("ba")) ;
                        jsonObject.put("channel", commonObject.getString("ch")) ;
                        jsonObject.put("is_new", commonObject.getString("is_new")) ;
                        jsonObject.put("model", commonObject.getString("md")) ;
                        jsonObject.put("os", commonObject.getString("os")) ;
                        jsonObject.put("uid", commonObject.getOrDefault("uid", "").toString()) ;
                        jsonObject.put("version_code", commonObject.getString("vc")) ;
                        // 3字段值
                        jsonObject.put("entry", startObject.getString("entry")) ;
                        jsonObject.put("loading_time", startObject.getLong("loading_time")) ;
                        jsonObject.put("open_ad_id", startObject.getLong("open_ad_id")) ;
                        jsonObject.put("open_ad_ms", startObject.getLong("open_ad_ms")) ;
                        jsonObject.put("open_ad_skip_ms", startObject.getLong("open_ad_skip_ms")) ;

                        jsonObject.put("ts", object.getLong("ts")) ;
                        // 4返回JSON字符串
                        return jsonObject.toJSONString();
                    }
                }
        );

        // 2.插入数据
        String[] columns = new String[]{
                "mid", "area_code","brand", "channel", "is_new", "model", "os", "uid", "version_code",
                "entry", "loading_time", "open_ad_id", "open_ad_ms", "open_ad_skip_ms",  "ts"
        };
        // 3.构造SQL
        String upsertSql = "INSERT INTO jtp_app_log.dwd_traffic_app_start_log(\n" +
                "    mid, area_code, brand, channel, is_new, model, os, uid, version_code,\n" +
                "    entry, loading_time, open_ad_id, open_ad_ms, open_ad_skip_ms,\n" +
                "    ts\n" +
                ") VALUES (\n" +
                "     ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?\n" +
                " )" ;

        // 4.执行
        JdbcUtil.sinkToClickhouseUpsertJson(jsonStream,columns, upsertSql);

    }

    // 曝光日志存储clickhouse
    public static void saveDisplayLog(DataStream<String> stream){

        SingleOutputStreamOperator<String> jsonStream = stream.map(
                new RichMapFunction<String, String>() {
                    @Override
                    public String map(String value) throws Exception {

                        // 1解析JSON
                        JSONObject object = JSON.parseObject(value);
                        JSONObject commonObject = object.getJSONObject("common");
                        JSONObject displayObject = object.getJSONObject("display");
                        JSONObject pageObject = object.getJSONObject("page");

                        // 2提取公共字段值
                        JSONObject jsonObject = new JSONObject();
                        jsonObject.put("mid", commonObject.getString("mid"));
                        jsonObject.put("area_code", commonObject.getString("ar"));
                        jsonObject.put("brand", commonObject.getString("ba"));
                        jsonObject.put("channel", commonObject.getString("ch"));
                        jsonObject.put("is_new", commonObject.getString("is_new"));
                        jsonObject.put("model", commonObject.getString("md"));
                        jsonObject.put("os", commonObject.getString("os"));
                        jsonObject.put("uid", commonObject.getOrDefault("uid", "").toString());
                        jsonObject.put("version_code", commonObject.getString("vc"));

                        // 3曝光日志字段值
                        jsonObject.put("display_type", displayObject.getOrDefault("display_type", ""));
                        jsonObject.put("display_item", displayObject.getOrDefault("item", ""));
                        jsonObject.put("display_item_type", displayObject.getOrDefault("item_type", ""));
                        jsonObject.put("pos_id", displayObject.getOrDefault("pos_id", ""));
                        jsonObject.put("order", displayObject.getOrDefault("order", ""));

                        // 4页面日志字段值
                        jsonObject.put("page_id", pageObject.getOrDefault("page_id", ""));
                        jsonObject.put("last_page_id", pageObject.getOrDefault("last_page_id", ""));
                        jsonObject.put("during_time", pageObject.getLong("during_time"));
                        jsonObject.put("sourceType", pageObject.getOrDefault("sourceType", ""));
                        jsonObject.put("page_item_type", pageObject.getOrDefault("item_type",""));
                        jsonObject.put("page_item", pageObject.getOrDefault("item", ""));

                        // 5时间戳
                        jsonObject.put("ts", object.getLong("ts"));

                        return jsonObject.toJSONString();
                    }
                }
        );

        // 2.插入数据
        String[] columns = new String[]{
                "mid", "area_code","brand", "channel", "is_new", "model", "os", "uid", "version_code",
                "display_type", "display_item", "display_item_type", "pos_id", "order",
                "page_id", "last_page_id", "during_time", "sourceType", "page_item_type", "page_item",
                "ts"
        };

        // 3.构造SQL
        String upsertSql = "INSERT INTO jtp_app_log.dwd_traffic_app_display_log(\n" +
                "    mid, area_code, brand, channel, is_new, model, os, uid,\n" +
                "    version_code, display_type, display_item_id, display_item_type,\n" +
                "    pos_id, order, page_id, last_page_id, during_time, sourceType,\n" +
                "    page_item_type, page_item, ts\n" +
                ") VALUES (\n" +
                "    ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?\n" +
                ")" ;

        // 4.执行
        JdbcUtil.sinkToClickhouseUpsertJson(jsonStream, columns, upsertSql);

    }

    // 动作日志存储clickhouse
    public static void saveActionLog(DataStream<String> stream){

        SingleOutputStreamOperator<String> jsonStream = stream.map(
                new RichMapFunction<String, String>() {
                    @Override
                    public String map(String value) throws Exception {

                        // 1解析JSON
                        JSONObject object = JSON.parseObject(value);
                        JSONObject commonObject = object.getJSONObject("common");
                        JSONObject actionObject = object.getJSONObject("action");
                        JSONObject pageObject = object.getJSONObject("page");

                        // 2提取公共字段值
                        JSONObject jsonObject = new JSONObject();
                        jsonObject.put("mid", commonObject.getString("mid"));
                        jsonObject.put("area_code", commonObject.getString("ar"));
                        jsonObject.put("brand", commonObject.getString("ba"));
                        jsonObject.put("channel", commonObject.getString("ch"));
                        jsonObject.put("is_new", commonObject.getString("is_new"));
                        jsonObject.put("model", commonObject.getString("md"));
                        jsonObject.put("os", commonObject.getString("os"));
                        jsonObject.put("uid", commonObject.getOrDefault("uid", "").toString());
                        jsonObject.put("version_code", commonObject.getString("vc"));

                        // 3动作日志字段值
                        jsonObject.put("action_id", actionObject.getOrDefault("action_id", ""));
                        jsonObject.put("action_item", actionObject.getOrDefault("item", ""));
                        jsonObject.put("action_item_type", actionObject.getOrDefault("item_type", ""));
                        jsonObject.put("action_ts",  actionObject.getLong("ts"));

                        // 4页面日志字段值
                        jsonObject.put("page_id", pageObject.getOrDefault("page_id", ""));
                        jsonObject.put("last_page_id", pageObject.getOrDefault("last_page_id", ""));
                        jsonObject.put("during_time", pageObject.getLong("during_time"));
                        jsonObject.put("sourceType", pageObject.getOrDefault("sourceType", ""));
                        jsonObject.put("page_item_type", pageObject.getOrDefault("item_type",""));
                        jsonObject.put("page_item", pageObject.getOrDefault("item", ""));

                        // 5时间戳
                        jsonObject.put("ts", object.getLong("ts"));

                        return jsonObject.toJSONString();
                    }
                }
        );

        // 2.插入数据
        String[] columns = new String[]{
                "mid", "area_code","brand", "channel", "is_new", "model", "os", "uid", "version_code",
                "action_id", "action_item", "action_item_type", "action_ts",
                "page_id", "last_page_id", "during_time", "sourceType", "page_item_type", "page_item",
                "ts"
        };

        // 3.构造SQL
        String upsertSql = "INSERT INTO jtp_app_log.dwd_traffic_app_action_log(\n" +
                "    mid, area_code, brand, channel, is_new, model, os, uid, version_code,\n" +
                "    action_id, action_item_id, action_item_type, action_ts, page_id,\n" +
                "    last_page_id, during_time, sourceType, page_item_type, page_item, ts\n" +
                ") VALUES (\n" +
                "    ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?\n" +
                ")" ;

        // 4.执行
        JdbcUtil.sinkToClickhouseUpsertJson(jsonStream, columns, upsertSql);

    }

    // 错误日志存储clickhouse
    public static void saveErrorLog(DataStream<String> stream) {
        SingleOutputStreamOperator<String> errorStream = stream.map(
                new RichMapFunction<String, String>() {
                    @Override
                    public String map(String values) throws Exception {
                        // 1解析JSON
                        JSONObject Object = JSON.parseObject(values);
                        JSONObject commonObject = Object.getJSONObject("common");
                        JSONObject errorObject = Object.getJSONObject("err");

                        JSONObject jsonObject = new JSONObject();
                        // todo common
                        jsonObject.put("mid", commonObject.getString("mid"));
                        jsonObject.put("area_code", commonObject.getString("ar"));
                        jsonObject.put("brand", commonObject.getString("ba"));
                        jsonObject.put("channel", commonObject.getString("ch"));
                        jsonObject.put("is_new", commonObject.getString("is_new"));
                        jsonObject.put("model", commonObject.getString("md"));
                        jsonObject.put("os", commonObject.getString("os"));
                        jsonObject.put("uid", commonObject.getOrDefault("uid", "").toString());
                        jsonObject.put("version_code", commonObject.getString("vc"));
                        // todo err
                        jsonObject.put("msg", errorObject.getString("msg").replace("\\", ""));
                        jsonObject.put("error_code", errorObject.getOrDefault("error_code", ""));

                        //todo page
                        if (null != Object.get("page")) {
                            JSONObject pageObject = Object.getJSONObject("page");
                            jsonObject.put("page_id", pageObject.getOrDefault("page_id", ""));
                            jsonObject.put("last_page_id", pageObject.getOrDefault("last_page_id", ""));
                            jsonObject.put("during_time", pageObject.getLong("during_time"));
                            jsonObject.put("sourceType", pageObject.getOrDefault("sourceType", ""));
                            jsonObject.put("item_type", pageObject.getOrDefault("item_type", ""));
                            jsonObject.put("item", pageObject.getOrDefault("item", ""));
                        } else {
                            jsonObject.put("page_id", "");
                            jsonObject.put("last_page_id", "");
                            jsonObject.put("during_time", "");
                            jsonObject.put("sourceType", "");
                            jsonObject.put("item_type", "");
                            jsonObject.put("item", "");
                        }

                        // todo start
                        if (null != Object.get("start")) {
                            JSONObject startObject = Object.getJSONObject("start");
                            jsonObject.put("entry", startObject.getString("entry"));
                            jsonObject.put("loading_time", startObject.getLong("loading_time"));
                            jsonObject.put("open_ad_id", startObject.getLong("open_ad_id"));
                            jsonObject.put("open_ad_ms", startObject.getLong("open_ad_ms"));
                            jsonObject.put("open_ad_skip_ms", startObject.getLong("open_ad_skip_ms"));
                        } else {
                            jsonObject.put("entry", "");
                            jsonObject.put("loading_time", "");
                            jsonObject.put("open_ad_id", "");
                            jsonObject.put("open_ad_ms", "");
                            jsonObject.put("open_ad_skip_ms", "");
                        }
                        // todo displays
//                        jsonObject.put("display_type",displaysObject.getString("display_type"));
                        if (null != Object.get("displays")) {
                            JSONArray displaysObject = Object.getJSONArray("displays");
                            jsonObject.put("displays", displaysObject.toJSONString());
                        } else {
                            jsonObject.put("displays", "");
                        }
                        // todo  actions
                        if (null != Object.get("actions")) {
                            JSONArray actionsObject = Object.getJSONArray("actions");
                            jsonObject.put("actions", actionsObject.toJSONString());
                        } else {
                            jsonObject.put("actions", "");
                        }
                        // 时间戳
                        jsonObject.put("ts", Object.getLong("ts"));
                        return jsonObject.toJSONString();
                    }
                }
        );

        String[] columns = new String[]{
                "mid", "area_code", "brand", "channel", "is_new", "model", "os", "uid", "version_code",
                "msg", "error_code", "page_id", "last_page_id", "during_time", "sourceType",
                "item_type", "item", "entry", "loading_time", "open_ad_id", "open_ad_ms",
                "open_ad_skip_ms", "displays", "actions", "ts"
        };

        String upsertSql = "INSERT INTO jtp_app_log.dwd_traffic_app_error_log (\n" +
                "    mid, area_code, brand, channel, is_new, model, os, uid, version_code,\n" +
                "    msg, error_code, page_id, last_page_id, during_time, sourceType,\n" +
                "    item_type, item, entry, loading_time, open_ad_id, open_ad_ms,\n" +
                "    open_ad_skip_ms, displays, actions, ts\n" +
                ") VALUES (\n" +
                "    ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?\n" +
                ")";

        JdbcUtil.sinkToClickhouseUpsertJson(errorStream, columns, upsertSql);

    }

}
